package com.parsifal.impl;

import com.parsifal.SerializableUtil;
import com.parsifal.iface.IClient;
import com.parsifal.callback.ClientCallback;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.parsifal.SerializableUtil.byteArrayToInt;
import static com.parsifal.SerializableUtil.intToByteArray;

/**
 * Created by yangming on 17-5-26.
 */
public class Client implements IClient {

    private static final int INT_SIZE = 4;

    private ClientCallback mClientCallback = null;

    private String mIP = null;

    private int mPort = 0;

    private boolean mClientRun = false;

    private SocketChannel mSocketChannel = null;

    /**
     * 接收数据的长度
     */
    private int mMessageLength = -1;

    /**
     * 携带接收数据的长度的缓冲区，4为int型数据字节数
     */
    private ByteBuffer mReceiveHeader = null;

    /**
     * 接收数据缓冲区
     */
    private ByteBuffer mReceiveBody = null;

    private ExecutorService mThreadPool = null;

    public static void main(String[] args) {
        final Client client = new Client();
        client.setIP("localhost");
        client.setPort(33497);
        client.setClientCallback(new ClientCallback() {
            @Override
            public void onConnected() {
                System.out.println("onConnected");
                client.send("你好");
            }

            @Override
            public void onDisconnected() {
                System.out.println("onDisconnected");
            }

            @Override
            public void onReceive(String message) {
                System.out.println("onReceive " + message);
            }
        });
        client.connect();
    }

    @Override
    public void setIP(String ip) {
        this.mIP = ip;
    }

    @Override
    public void setPort(int port) {
        this.mPort = port;
    }

    @Override
    public void connect() {
        mThreadPool = Executors.newFixedThreadPool(2);
        mThreadPool.execute(new Runnable() {
            @Override
            public void run() {
                mReceiveHeader = ByteBuffer.allocate(INT_SIZE);
                InetSocketAddress SERVER_ADDRESS = new InetSocketAddress(mIP, mPort);
                Selector selector = null;
                SocketChannel socketChannel = null;
                try {
                    selector = Selector.open();
                    socketChannel = SocketChannel.open();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector, SelectionKey.OP_CONNECT);
                    socketChannel.connect(SERVER_ADDRESS);
                    mClientRun = true;
                    while (mClientRun) {
                        selector.select();
                        Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                        while (it.hasNext()) {
                            SelectionKey selectionKey = it.next();
                            if (selectionKey.isConnectable()) {
                                handleConnect(selectionKey);
                            }
                            if (selectionKey.isReadable()) {
                                handleRead(selectionKey);
                            }
                            it.remove();
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                } finally {
                    mClientRun = false;
                    try {
                        if (null != selector) {
                            selector.close();
                        }
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                    try {
                        if (null != socketChannel) {
                            socketChannel.close();
                        }
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }
        });
    }

    @Override
    public void setClientCallback(ClientCallback callback) {
        this.mClientCallback = callback;
    }

    private void handleConnect(SelectionKey key) {
        mSocketChannel = (SocketChannel) key.channel();
        try {
            if (mSocketChannel.isConnectionPending()) {
                mSocketChannel.finishConnect();
                key.interestOps(SelectionKey.OP_READ);
                if (null != mClientCallback) {
                    mClientCallback.onConnected();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void handleRead(SelectionKey key) {
        mSocketChannel = (SocketChannel) key.channel();
        try {
            if (mMessageLength == -1) {
                long bytesRead = mSocketChannel.read(mReceiveHeader);
                if (-1 == bytesRead) {
                    if (null != mClientCallback) {
                        mClientCallback.onDisconnected();
                    }
                    mSocketChannel.close();
                }
                //并且缓存区的长度大于4(包头部分已经接受完毕)
                if (!mReceiveHeader.hasRemaining()) {
                    mMessageLength = byteArrayToInt(mReceiveHeader.array());
                    System.out.println("dataSize ......" + mMessageLength);
                    mReceiveHeader.clear();
                    mReceiveBody = ByteBuffer.allocate(mMessageLength);
                }
            } else {
                // 尝试读取数据区域
                long bytesRead = mSocketChannel.read(mReceiveBody);
                if (-1 == bytesRead) {
                    if (null != mClientCallback) {
                        mClientCallback.onDisconnected();
                    }
                    mSocketChannel.close();
                }
                if (!mReceiveBody.hasRemaining()) {
                    // 这个时候可以解析数据
                    String message = SerializableUtil.byteArrayToString(mReceiveBody.array());
                    if (null != mClientCallback) {
                        mClientCallback.onReceive(message);
                    }
                    mReceiveBody.clear();
                    mMessageLength = -1;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void send(final String pMsg) {
        if (null != pMsg) {
            if (!mThreadPool.isShutdown()) {
                mThreadPool.execute(new Runnable() {
                    @Override
                    public void run() {
                        byte[] messageBytes = SerializableUtil.stringToByteArray(pMsg);
                        int count = messageBytes.length;
                        ByteBuffer sendBuffer = ByteBuffer.allocate(messageBytes.length + INT_SIZE);

                        try {
                            sendBuffer.clear();
                            sendBuffer.put(intToByteArray(count));
                            sendBuffer.put(messageBytes);
                            sendBuffer.flip();
                            mSocketChannel.write(sendBuffer);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
            }
        } else {
            System.out.println("message is null");
        }
    }

    @Override
    public void stopClient() {
        mClientRun = false;
        if (null != mThreadPool) {
            mThreadPool.shutdownNow();
        }
    }
}
